/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.queue.impl;

import java.util.Arrays;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Array-backed non-blocking queue.
 * 
 * @author Alex Antonau
 * */
public final class ArrayDataQueue<E> {

    /** The queued items  */
    private final E[] items;
    /** items index for next take, poll or remove */
    private int takeIndex;
    /** items index for next put, offer, or add. */
    private int putIndex;
    /** Number of items in the queue */
    private volatile int count;
    /** Main lock guarding all access */
    private final ReentrantLock lock;

    /**
     * Inserts element at current put position.
     * Call only when holding lock.
     */
    private void insert(E x) {
        items[putIndex++] = x;
        if(putIndex == items.length) putIndex = 0;
        ++count;
    }
    /**
     * Inserts elements at current put position.
     * Call only when holding lock.
     */
    private void insert(E[] elems, int takeIndex, int size) {
    	final int left = items.length - putIndex;
    	if(left < size){
        	System.arraycopy(elems, takeIndex, items, putIndex, left);
        	putIndex = size - left;
        	System.arraycopy(elems, takeIndex+left, items, 0, putIndex);
    	}else{
        	System.arraycopy(elems, takeIndex, items, putIndex, size);
        	putIndex += size;
    	}
    	count+=size;
        if(putIndex == items.length) putIndex = 0;
    }

    /**
     * Extracts element at current take position.
     * Call only when holding lock.
     */
    private E extract() {
        final E[] items = this.items;
        E x = items[takeIndex];
        items[takeIndex++] = null;
        if(takeIndex == items.length) takeIndex = 0;
        --count;
        return x;
    }

    /**
     * Creates an <tt>ArrayDataQueue</tt> with the given (fixed)
     * capacity.
     *
     * @param capacity the capacity of this queue
     * @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
     */
	@SuppressWarnings("unchecked")
	public ArrayDataQueue(int capacity) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = (E[]) new Object[capacity];
        lock = new ReentrantLock();
	}

	public boolean tryAdd(E e) {
        if (e == null) throw new NullPointerException();
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        if(lock.tryLock()){
	        try {
	        	if(count == items.length) return false;
	            insert(e);
	            return true;
	        } finally {
	            lock.unlock();
	        }
        }
        return false;
	}
	public void add(E e) throws IllegalStateException {
        if (e == null) throw new NullPointerException();
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        	if(count == items.length) throw new IllegalStateException("Queue overflow");
            insert(e);
        } finally {
            lock.unlock();
        }
    }
	public void addFromTail(ArrayDataQueue<E> queue, int size) throws IllegalStateException {
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        	if(count + size > items.length) throw new IllegalStateException();
        	final ReentrantLock otherLock = queue.lock;
        	otherLock.lock();
        	try {
        		size = Math.min(size, queue.count);
        		if(queue.putIndex >= size){
        			queue.putIndex -= size;
        			insert(queue.items, queue.putIndex, size);
        			Arrays.fill(queue.items, queue.putIndex, queue.putIndex + size, null);
        		}else{
        			insert(queue.items, 0, queue.putIndex);
        			Arrays.fill(queue.items, 0, queue.putIndex, null);
        			final int left = size - queue.putIndex;
        			queue.putIndex = queue.items.length - left;
        			insert(queue.items, queue.putIndex , left);
        			Arrays.fill(queue.items, queue.putIndex, queue.items.length, null);
        		}
    			queue.count -= size;
        	} finally {
        		otherLock.unlock();
        	}
        } finally {
            lock.unlock();
        }
    }
	
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == 0) return null;
            E x = extract();
            return x;
        } finally {
            lock.unlock();
        }
    }
    public E tryPoll() {
        final ReentrantLock lock = this.lock;
        if(lock.tryLock()){
	        try {
	            if (count == 0) return null;
	            E x = extract();
	            return x;
	        } finally {
	            lock.unlock();
	        }
        }
        return null;
    }
    
    public int size() {
    	return count;
    }
    
    
	public int capacity() {
		return items.length;
	}

	public boolean isFull(){
		return count == items.length;
    }
    /**
     * Atomically removes all of the elements from this queue.
     * The queue will be empty after this call returns.
     */
    public void clear() {
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int k = count;
            while (k-- > 0) {
                items[i++] = null;
                if(i==items.length) i = 0;
            }
            count = 0;
            putIndex = 0;
            takeIndex = 0;
        } finally {
            lock.unlock();
        }
    }

	public boolean isEmpty() {
		return count==0;
	}


	@Override
	public String toString() {
		return "ArrayDataQueue [capacity=" + items.length + ", count=" + count + "]";
	}

	/**
	 * Returns snapshot of all elements currently resided in the queue.
	 * */
	public Object[] toArray(){
		Object[] res = null;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final int size = count;
    		res = new Object[size];
            int i = takeIndex;
            int k = 0;
            while (k < size) {
                res[k++] = items[i++];
                if(i==items.length) i = 0;
            }
        } finally {
            lock.unlock();
        }
        return res;
	}

}
